package com.mrbeard.greenbroswer.utils;


import com.mrbeard.greenbroswer.entites.Log;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.QueueingConsumer;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.TimeoutException;


/**
 * Created by 胡彬 on 2018/4/13.
 */
public class RabbitmqUtil {

    /**
     * 队列名称
     */
    private String exchangeName;
    /**
     * 信息字符串
     */
    private String logMessag;
    private String messageJson;

    /**
     * 主机地址
     */
    private String hostUrl = null;
    private Consumer consumer = null;
    /**
     * 返回信息内容
     */
    private String mes = null;
    private String type = null;
    private ConnectionFactory factory = new ConnectionFactory();

    private String key = null;
    private String queueName = null;
//    /**
//     *
//     * @param exchangeName
//     * @param hostUrl
//     * @param type
//     */
//    public RabbitmqUtil(String exchangeName, String hostUrl,String type) {
//        this.exchangeName = exchangeName;
//        this.hostUrl = hostUrl;
//        this.type = type;
//    }

    /**
     * @param exchangeName
     * @param queueName
     * @param hostUrl
     * @param type
     * @param key
     */
    public RabbitmqUtil(String exchangeName, String queueName, String hostUrl, String type, String key) {
        this.exchangeName = exchangeName;
        this.hostUrl = hostUrl;
        this.type = type;
        this.key = key;
        this.queueName = queueName;
    }


    /**
     *
     * @param exchangeName
     * @param queueName
     * @param hostUrl
     * @param type
     * @param key
     */
    public RabbitmqUtil(String exchangeName, String queueName, String hostUrl, String type, String key,String logMessag) {
        this.exchangeName = exchangeName;
        this.hostUrl = hostUrl;
        this.type = type;
        this.key = key;
        this.queueName = queueName;
        this.logMessag = logMessag;
    }


//    /**
//     *
//     * @param exchangeName
//     * @param hostUrl
//     * @param messageJson
//     * @param type
//     */
//    public RabbitmqUtil(String exchangeName, String hostUrl, String messageJson,String type) {
//        this.exchangeName = exchangeName;
//        this.messageJson = messageJson;
//        this.hostUrl = hostUrl;
//        this.type = type;
//    }


    public String receive() throws Exception {
        if (type.equals("fanout")) {
            factory.setHost(hostUrl);
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            channel.exchangeDeclare(exchangeName, type);
            String queueName = channel.queueDeclare().getQueue();
            channel.queueBind(queueName, exchangeName, "");

            if (consumer == null) {
                consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException {
                        String message = new String(body, "UTF-8");
                        mes = message;
                        System.out.println(" [x] Received '" + message + "'");
                    }
                };
            }
            channel.basicConsume(queueName, true, consumer);
            if (mes != null) {
                channel.close();
                connection.close();
                return mes;
            } else {
                channel.basicConsume(queueName, true, consumer);
            }
            return mes;
        } else if (type.equals("direct")) {
            //1.打开连接和创建频道，与发送端一样
            factory.setHost(hostUrl);
            factory.setRequestedHeartbeat(2);//是否断网
            //创建一个连接   创建一个频道
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            //2.声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
            channel.queueDeclare(exchangeName, true, false, false, null);

            //3.创建队列消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(exchangeName, true, consumer);//指定消费队列
            while (true){
                //4.开启nextDelivery阻塞方法（内部实现其实是阻塞队列的take方法）
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                return  message;
            }
        } else if (type.equals("topic")) {
            factory.setHost(hostUrl);
            factory.setRequestedHeartbeat(2);//是否断网
            // 获取到连接以及mq通道
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(queueName, true, false, false, null);
            // 绑定队列到交换机
            channel.queueBind(queueName, exchangeName, key);
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(queueName, false, consumer);
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                mes = message;
                Thread.sleep(10);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                if(mes != null){
                    return mes;
                }
            }
        }
        return mes;
    }



    /**
     * 发送log
     * @throws Exception
     */
    public void send(){
        try {
            factory.setHost(hostUrl);
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            channel.exchangeDeclare(exchangeName, "topic");
            channel.queueDeclare(queueName,true,false,false,null);
            channel.queueBind(queueName,exchangeName,key);
            channel.basicPublish(exchangeName,key,null,logMessag.getBytes());
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}
